package com.tang;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.ArrayList;
import java.util.List;

/**
 * 全局有序生产者
 */
public class PartlyProducer {
    public static void main(String[] args) throws Exception {
        //第二个参数开启消息追踪
        DefaultMQProducer producer = new DefaultMQProducer("producer_group", true);
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();
        List<Order> list1 = buildOrderList(1, "Buy", "Pay", "Finish");
        List<Order> list2 = buildOrderList(2, "Buy", "Pay");
        List<Order> list3 = buildOrderList(3, "Pay", "Finish");
        List<Order> orderList = new ArrayList<>();
        orderList.addAll(list1);
        orderList.addAll(list2);
        orderList.addAll(list3);
        int size = orderList.size();
        for (int i = 0; i < size; i++) {
            Order order = orderList.get(i);
            String info = order.toString();
            //这里的主题是局部有序的主题
            Message msg = new Message("Partly_Orderly-Topic", "globalTag", info.getBytes());
            //setKeys这个方法就是设置每个消息的MessageKey
            msg.setKeys("Partly_Orderly-Topic");
            producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message message, Object o) {
                    //这个参数o就是我们在send方法里面写的第三个参数的orderId的值，通过取模运算可以让让相同的订单的id都进入到同一个MessageQueue里面
                    //返回的值就是哪一个MessageQueue
                    int orderId = (int) o;
                    int idx = orderId % mqs.size();
                    return mqs.get(idx);
                }
            }, order.orderId);
        }
        System.out.println("发送成功");
    }

    public static List<Order> buildOrderList(int orderId, String... descList) {
        List<Order> resultList = new ArrayList<>();
        for (String desc : descList) {
            Order order = new Order();
            order.orderId = orderId;
            order.desc = desc;
            resultList.add(order);
        }
        return resultList;
    }

    public static class Order {
        /**
         * 订单id
         */
        int orderId;
        /**
         * 订单描述
         */
        String desc;

        @Override
        public String toString() {
            return "Order{" +
                    "orderId=" + orderId +
                    ", desc='" + desc + '\'' +
                    '}';
        }
    }
}